package com.zyh.flink.day02.source;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;

public class FromKafkaTest2 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        /*
        * flink要作为kafka的消费者,设置4个参数创建出kafka消费对象
        * */
        String topicName = "topic-flink-source";
        //反序列化方式,将kafka中存储的消息的value部分反序列化回String
        JSONKeyValueDeserializationSchema jsonKeyValueDeserializationSchema = new JSONKeyValueDeserializationSchema(true);
        Properties props = new Properties();
        //kafka集群入口
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"kafka24:9092");
        //消费者组名
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"flink");
        //创建flinkKafka消费对象
        FlinkKafkaConsumer<ObjectNode> consumer = new FlinkKafkaConsumer<>(topicName,jsonKeyValueDeserializationSchema,props);

        DataStreamSource<ObjectNode> result = environment.addSource(consumer);

        SingleOutputStreamOperator<String> map = result.map(new MapFunction<ObjectNode, String>() {
            @Override
            public String map(ObjectNode jsonNodes) throws Exception {
                /*
                * objectNode:
                *   key,value,metadata
                * */
                JsonNode value = jsonNodes.get("value");
                String name = value.get("name").asText();
                int age = value.get("age").asInt();

                return name+","+age;
            }
        });

        map.print();
        environment.execute("a");
    }
}